package com.uestc.coprocessor.userfeedback;

import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.util.Bytes;

import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.Service;
import com.uestc.coprocessor.userfeedback.UserFeedbackProtocol.CountRequest;
import com.uestc.coprocessor.userfeedback.UserFeedbackProtocol.CountResponse;

public class UserFeedbackAggregation extends UserFeedbackProtocol.UserFeedbackService
		implements Coprocessor, CoprocessorService {

	private RegionCoprocessorEnvironment env;

	@Override
	public void getUserFeedback(RpcController controller, CountRequest request, RpcCallback<CountResponse> done) {
		Scan scan = new Scan();
		String[] types = request.getType().split(";"); // 问题和要查看的题目以分号隔开
		if (types.length < 2)
			return;
		String sheet = types[0]; // 问卷种类；
		String questions = types[1]; // 要统计的问题，以逗号隔开；
		String[] question = questions.split(",");
		Map<String, Long> trueMap = new LinkedHashMap<>();
		Map<String, Long> falseMap = new LinkedHashMap<>();
		for (String q : question) {
			scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes(q));
			trueMap.put(q, 0l);
			falseMap.put(q, 0l);
		}
		scan.setRowPrefixFilter(Bytes.toBytes(sheet + ";"));
		UserFeedbackProtocol.CountResponse response = null;
		InternalScanner scanner = null;

		try {
			scanner = env.getRegion().getScanner(scan);
			List<Cell> results = new ArrayList<Cell>();
			boolean hasMore = false;
			do {
				hasMore = scanner.next(results);
				for (Cell c : results) {
					String qualifier = Bytes.toString(CellUtil.cloneQualifier(c));
					String answer = Bytes.toString(CellUtil.cloneValue(c));
					if ("true".equals(answer)) {
						Long oldValue = trueMap.get(qualifier);
						if (oldValue == null)
							trueMap.put(qualifier, 0l);
						else
							trueMap.put(qualifier, oldValue + 1l);
					} else if ("false".equals(answer)) {
						Long oldValue = falseMap.get(qualifier);
						if (oldValue == null)
							falseMap.put(qualifier, 0l);
						else
							falseMap.put(qualifier, oldValue + 1l);
					}
				}
				results.clear();
			} while (hasMore);
			StringBuilder sb1 = new StringBuilder();
			StringBuilder sb2 = new StringBuilder();
			for (Entry<String, Long> e : trueMap.entrySet()) {
				sb1.append(e.getKey());
				sb1.append(":");
				sb1.append(e.getValue());
				sb1.append(";");
			}
			for (Entry<String, Long> e : falseMap.entrySet()) {
				sb2.append(e.getKey());
				sb2.append(":");
				sb2.append(e.getValue());
				sb2.append(";");
			}
			sb1.append("&");
			sb1.append(sb2);
			response = UserFeedbackProtocol.CountResponse.newBuilder().setResult(sb1.toString()).build();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if (scanner != null) {
				try {
					scanner.close();
				} catch (IOException ignored) {
				}
			}
		}
		done.run(response);
	}

	@Override
	public Service getService() {
		return this;
	}

	@Override
	public void start(CoprocessorEnvironment arg0) throws IOException {
		if (arg0 instanceof RegionCoprocessorEnvironment) {
			this.env = (RegionCoprocessorEnvironment) arg0;
		} else {
			throw new CoprocessorException("Must be loaded on a table region!");
		}
	}

	@Override
	public void stop(CoprocessorEnvironment arg0) throws IOException {

	}

}
